package com.intellif.mozping.producer;


import com.intellif.mozping.util.KafkaHandler;
import com.intellif.mozping.util.TaskUtil;
import com.intellif.mozping.util.Util;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.util.Properties;
import java.util.Random;

@Component
public class TestKafka {

    public static final Logger LOG = LoggerFactory.getLogger(TestKafka.class);
    private static String path = "./src/main/resource/system.properties";
    private static final Random random = new Random();

    @Value("${message.num}")
    private int messageNum = 10 * 10000;

    @Value("${count.batch}")
    private int countBatch = 10000;

    @Value("${message.size}")
    private int messageSize = 3000;

    @Value("${cluster.topics}")
    private String topics;

    @Autowired
    TaskUtil taskUtil;

    @Async
    public void send() throws InterruptedException {
        Properties properties = taskUtil.getKafkaConf();
        KafkaHandler handler = KafkaHandler.getInstance(properties);
        //获取topic列表和消息内容
        String testMassage = Util.getTestMsg();
        String[] topicList = topics.split(",");

        long begin = System.currentTimeMillis();
        long lastTimeStamp = System.currentTimeMillis() - 1;
        for (int i = 0; i < messageNum; i++) {
            handler.handle(topicList[i % topicList.length], testMassage);
            if (i % countBatch == 0 && i > 0) {
                long interval = System.currentTimeMillis() - lastTimeStamp;

                double speed = countBatch / (interval / 1000.0);
                LOG.info("发送第 " + i + " 条数据完毕! ,最近" + countBatch + "条平均发送速率为：" + (int) speed + "条/S ");
                lastTimeStamp = System.currentTimeMillis();
            }
        }
        long total = System.currentTimeMillis() - begin;
        LOG.info("发送" + messageNum + "条数据完毕，总耗时" + total + " ms ,每秒发送条数" + (messageNum / (total / 1000)));
    }
}
